package cn.hy.dataSync.job.scheduler;


import cn.hy.dataSync.config.CoreConstant;
import cn.hy.dataSync.config.TaskConfig;
import cn.hy.dataSync.job.State;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import cn.hy.dataSync.statistics.container.communicator.AbstractContainerCommunicator;
import cn.hy.dataSync.task.ErrorRecordChecker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.Validate;

import java.util.List;


/**
 * 调度器
 */
@Slf4j
public abstract class AbstractScheduler {

    private ErrorRecordChecker errorLimit;

    private AbstractContainerCommunicator containerCommunicator;

    private String jobId;

    public String getJobId() {
        return jobId;
    }

    public AbstractScheduler(AbstractContainerCommunicator containerCommunicator) {
        this.containerCommunicator = containerCommunicator;
    }

    public void schedule(final List<TaskConfig> taskConfigs) {
        Validate.notNull(taskConfigs, "scheduler 配置不能为空");
        int jobReportIntervalInMillSec = taskConfigs.get(0).getCoreConfig()
                .getInteger(CoreConstant.JOB_REPORT_INTERVAL, 30000);
        int jobSleepIntervalInMillSec = taskConfigs.get(0).getCoreConfig()
                .getInteger(CoreConstant.JOB_SLEEP_INTERVAL, 10000);

        this.jobId = taskConfigs.get(0).getJobConfig().getJobId();

        this.errorLimit = new ErrorRecordChecker(taskConfigs.get(0).getJobConfig().getErrorLimit());

        this.containerCommunicator.registerCommunication(taskConfigs);

        int totalTasks = calculateTaskCount(taskConfigs);

        //启动所有执行的任务
        startAllTask(taskConfigs);

        Communication lastJobContainerCommunication = new Communication();

        long lastReportTimeStamp = System.currentTimeMillis();
        try {
            while (true) {
                /*
                 * 1：搜集任务状态
                 * 2：获取上报信息并上报
                 * 3：错误及限制检查
                 * 4：成功状态处理
                 * 5：Killing 状态处理
                 * 6：失败状态处理
                 * 7：刷新任务状态，暂短睡眠后开始下一次循环
                 */
                Communication nowJobContainerCommunication = this.containerCommunicator.collect();
                nowJobContainerCommunication.setTimestamp(System.currentTimeMillis());
                log.debug(nowJobContainerCommunication.toString());

                //汇报周期
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > jobReportIntervalInMillSec) {
                    Communication reportCommunication = CommunicationTool
                            .getReportCommunication(nowJobContainerCommunication, lastJobContainerCommunication, totalTasks);

                    this.containerCommunicator.report(reportCommunication);
                    lastReportTimeStamp = now;
                    lastJobContainerCommunication = nowJobContainerCommunication;
                }

                errorLimit.checkRecordLimit(nowJobContainerCommunication);

                if (nowJobContainerCommunication.getState() == State.SUCCEEDED) {
                    log.info("调度器完成了所有任务");
                    break;
                }

                if (isJobKilling(this.getJobId())) {
                    dealKillingStat(this.containerCommunicator, totalTasks);
                } else if (nowJobContainerCommunication.getState() == State.FAILED) {
                    dealFailedStat(this.containerCommunicator, nowJobContainerCommunication.getThrowable());
                }

                Thread.sleep(jobSleepIntervalInMillSec);
            }
        } catch (InterruptedException e) {
            log.error("捕获到InterruptedException异常!", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 启动所有任务
     *
     * @param taskConfigs 任务配置
     */
    protected abstract void startAllTask(List<TaskConfig> taskConfigs);

    /**
     * 失败状态处理
     *
     * @param frameworkCollector 状态搜集器
     * @param throwable          异常信息
     */
    protected abstract void dealFailedStat(AbstractContainerCommunicator frameworkCollector, Throwable throwable);

    /**
     * killing 状态处理
     *
     * @param frameworkCollector 状态搜集器
     * @param totalTasks         任务数
     */
    protected abstract void dealKillingStat(AbstractContainerCommunicator frameworkCollector, int totalTasks);

    /**
     * 计算任务数
     *
     * @param taskConfigs 任务配置
     * @return 任务数
     */
    private int calculateTaskCount(List<TaskConfig> taskConfigs) {
        return taskConfigs.size();
    }

    /**
     * 任务是否处于 killing 状态
     *
     * @param jobId 任务ID
     * @return true：是，false：否
     */
    protected abstract boolean isJobKilling(String jobId);
}
